/**
 * <p>文件名:		BaseProducer.java</p>
 * <p>版权:		CopyrightTag</p>
 * <p>公司:		CompanyTag</p>
 * @author		周华彬(zhouhuabin@ctfo.com, zhou_hua_bin@163.com)
*/

package com.caits.lbs.framework.services.kafka;

import java.util.Collection;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.apache.log4j.Logger;

import com.caits.lbs.framework.log.CommonLogFactory;
/**
 * <p>BaseProducer</p>
 * <p>TODO</p>
 *
 * @author		周华彬(zhouhuabin@ctfo.com, zhou_hua_bin@163.com)
 * @version		0.0.0
 * <table style="border:1px solid gray;">
 * <tr>
 * <th width="100px">版本号</th><th width="100px">动作</th><th width="100px">修改人</th><th width="100px">修改时间</th>
 * </tr>
 * <!-- 以 Table 方式书写修改历史 -->
 * <tr>
 * <td>0.0.0</td><td>创建类</td><td>zhouhuabin</td><td>2015年4月13日 下午5:44:17</td>
 * </tr>
 * <tr>
 * <td>XXX</td><td>XXX</td><td>XXX</td><td>XXX</td>
 * </tr>
 * </table>
 */

public class BaseProducer {  
	/** 日志记录器 */
	private Logger log = CommonLogFactory.getLog();
    /** 消息生产对象 */
    private Producer<String,String> procuder;  
    /**
     * 构造器
     * 自动加载classpath:producer.properties文件
     * @throws Exception 
     */
    public BaseProducer() throws Exception{  
        Properties properties = new Properties();  
        properties.load(ClassLoader.getSystemResourceAsStream("producer.properties"));  
        ProducerConfig config = new ProducerConfig(properties);
        procuder = new Producer<String, String>(config);  
    }  
  
      
    /**
     * 发送单个消息 
     * @param topicName
     * @param message
     */
    public void send(String topicName,String message) {  
        if(topicName == null || message == null){  
            return;  
        }  
        KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,"2",message);
        //如果具有多个partitions,请使用new KeyedMessage(String topicName,K key,V value).  
        procuder.send(km);
        log.info("send message succeed."+message); 
    }  
      
    /**
     * 发送一组消息 
     * @param topicName
     * @param messages
     */
    public void send(String topicName,Collection<String> messages) {  
        if(topicName == null || messages == null||messages.isEmpty()){  
            return;  
        }  
        for(String entry : messages){  
        	send(topicName,entry);
        }  
    }  
      
    /**
     * 关闭连接 
     */
    public void close(){  
        procuder.close();  
    }  
      
  
}  
